package ecarx.com.cn.ecarxproject;


import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManagerFactory;
import java.io.BufferedInputStream;
import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.security.KeyStore;
import java.security.cert.Certificate;
import java.security.cert.CertificateException;
import java.security.cert.CertificateFactory;
import java.util.Map;

import ecarx.com.cn.ecarxproject.util.Base64Util;
import ecarx.com.cn.ecarxproject.util.Config;
import ecarx.com.cn.ecarxproject.util.IotAuthUtil;
import ecarx.com.cn.ecarxproject.util.Md5;

/**
 * 类描述:
 * 16/4/12.
 */
public class MQTTClient {
	
    public static void main() throws Exception {

        //Step 1: 获取配置信息
        System.out.println("开始获取配置信息!");
        if (Config.deviceName.equals(Config.deviceName2))
        {
        	System.out.println("设备1和设备2不能相同!");
        	return;
        }
        Map<String, String> result = IotAuthUtil.auth(Config.deviceName, Config.deviceSecret);
		
        //1.1 得到公钥的BASE64编码以后的字符串数据
        String pubKey = result.get("pubkey");
		
        //1.2 得到连接的目的地IP与端口
        String servers = result.get("servers");
        String targetServer = servers.substring(0, servers.indexOf("|"));

        //1.3 得到BASE64字符串解码以后的公钥证书文件
        byte[] pubKeyByteContent = Base64Util.decode(pubKey);
        
        //1.4 得到设备的ID
        String deviceId1 = result.get("deviceId");
        result = IotAuthUtil.auth(Config.deviceName2, Config.deviceSecret2);
        String deviceId2 = result.get("deviceId");

        System.out.println("获取配置成功! 接入地址: " + targetServer);

        //Step 2: 设备1接入并订阅Topic
        deviceConnectAndSub(pubKeyByteContent, targetServer, deviceId1);

        //Step 3: 设备2接入并发布消息到Topic中
        deviceConnectAndPub(pubKeyByteContent, targetServer, deviceId2);
    }
    
    private static void deviceConnectAndPub(byte[] stest, String targetServer, String deviceId) throws Exception {
        System.out.println("==================================================");
        System.out.println("设备:" + Config.deviceName2 + "开始进行连接……");
        System.out.println("==================================================");
        final String topic = Config.topic;

        String broker = "ssl://" + targetServer;
        //客户端ID格式: productKey + deviceId.
        String clientId = Config.productKey + ":" + deviceId;

        MemoryPersistence persistence = new MemoryPersistence();
        SSLSocketFactory socketFactory = createSSLSocket(stest);
        try {
            final MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setMqttVersion(4);// MQTT 3.1.1
            connOpts.setSocketFactory(socketFactory);
            String signUserName = signUserName4Device2(deviceId);
            connOpts.setUserName(signUserName);
            connOpts.setKeepAliveInterval(65);
            System.out.println(Config.deviceName2 + "进行连接, 目的地: " + broker);
            sampleClient.connect(connOpts);
            sampleClient.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    System.out.println("连接失败,原因:" + cause);
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    System.out.println("接收到消息,来至Topic [" + topic + "] , 内容是:[" + new String(message.getPayload(), "UTF-8") + "]");
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("消息发送成功! " + token.getResponse().getKey());
                }
            });
            String content = "Message From Device:" + Config.deviceName2;
            MqttMessage message = new MqttMessage(content.getBytes());
            message.setQos(1);
            sampleClient.publish(topic, message);
            System.out.println("消息发布成功!");
        } catch (MqttException me) {
            System.out.println("MQTT连接异常: 异常码 " + me.getReasonCode() + ". 消息:" + me.getMessage() + " . 具体原因:" + me.getCause() + ". 异常:"
                    + me);
        }
    }

    private static void deviceConnectAndSub(byte[] stest, String targetServer, String deviceId) throws Exception {
        System.out.println("==================================================");
        System.out.println("设备:" + Config.deviceName + "开始进行连接……");
        System.out.println("==================================================");
        final String topic = Config.topic;
		
        String broker = "ssl://" + targetServer;
        //客户端ID格式: productKey + deviceId.
        String clientId = Config.productKey + ":" + deviceId;
		
        MemoryPersistence persistence = new MemoryPersistence();
        SSLSocketFactory socketFactory = createSSLSocket(stest);
        try {
            final MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setMqttVersion(4);// MQTT 3.1.1
            connOpts.setSocketFactory(socketFactory);
            String signUserName = signUserName(deviceId);
            connOpts.setUserName(signUserName);
            connOpts.setKeepAliveInterval(65);
            System.out.println("进行连接, 目的地: " + broker);
            sampleClient.connect(connOpts);
            sampleClient.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    System.out.println("连接失败,原因:" + cause);
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    System.out.println("接收到消息,来至Topic [" + topic + "] , 内容是:[" + new String(message.getPayload(), "UTF-8") + "]");
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("消息发送成功! " + token.getResponse().getKey());
                }
            });
            System.out.println("订阅：" + topic);
            sampleClient.subscribe(topic);

            System.out.println("订阅成功! 等待Topic消息中……! 10秒以后断开连接!");
            /*new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(10000);
                        sampleClient.disconnect();
                        System.out.println("客户端ID：" + sampleClient.getClientId() + "断开连接成功!");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }).start();*/
        } catch (MqttException me) {
            System.out.println("MQTT连接异常: 异常码 " + me.getReasonCode() + ". 消息:" + me.getMessage() + " . 具体原因:" + me.getCause() + ". 异常:"
                    + me);
        }
    }

    /**
     * 用户名设置规则: ToUpperCase(MD5_32(productKey+productSecret+deviceId+deviceSecret))
     * @param deviceId 
     *
     * @return
     */
    private static String signUserName(String deviceId) {
        String sign = Config.productKey + Config.productSecret + deviceId + Config.deviceSecret;
        String signUserName = Md5.getInstance().md5_32(sign).toUpperCase();
        return signUserName;
    }

    private static String signUserName4Device2(String deviceId) {
    	String sign = Config.productKey + Config.productSecret + deviceId + Config.deviceSecret2;
        String signUserName = Md5.getInstance().md5_32(sign).toUpperCase();
        return signUserName;
    }

    private static SSLSocketFactory createSSLSocket(byte[] pubKeyByteContent) throws Exception {
        InputStream is = new ByteArrayInputStream(pubKeyByteContent);
        InputStream caInput = new BufferedInputStream(is);
        CertificateFactory cf = CertificateFactory.getInstance("X.509");
        Certificate ca = null;
        try {
            ca = cf.generateCertificate(caInput);
        } catch (CertificateException e) {
            e.printStackTrace();
        } finally {
            caInput.close();
        }
        String keyStoreType = KeyStore.getDefaultType();
        KeyStore keyStore = KeyStore.getInstance(keyStoreType);
        keyStore.load(null, null);
        keyStore.setCertificateEntry("ca", ca);
        String tmfAlgorithm = TrustManagerFactory.getDefaultAlgorithm();
        TrustManagerFactory tmf = TrustManagerFactory.getInstance(tmfAlgorithm);
        tmf.init(keyStore);
        SSLContext context = SSLContext.getInstance("TLSV1.2");
        context.init(null, tmf.getTrustManagers(), null);
        SSLSocketFactory socketFactory = context.getSocketFactory();
        return socketFactory;
    }

}
